package com.stone.tech.cdc.listener;

import com.stone.tech.cdc.model.ChangeRow;
import com.stone.tech.cdc.model.Message;
import com.stone.tech.cdc.utils.JSONUtils;
import io.debezium.engine.ChangeEvent;
import io.debezium.engine.DebeziumEngine;
import io.debezium.engine.format.Json;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.stereotype.Component;

import java.util.*;
import java.util.concurrent.Executors;

/**
 * @author fujianjian
 * @date 2023/3/1 22:11
 **/
@Slf4j
@Component
public class DebeziumMysqlListener implements InitializingBean, DisposableBean {

    private final List<DebeziumEngine<ChangeEvent<String, String>>> engineList = new ArrayList<>();

    public DebeziumMysqlListener(@Qualifier("debeziumMysqlProperties") List<Properties> list) {
        for (Properties props : list) {
            this.engineList.add(DebeziumEngine.create(Json.class)
                    .using(props)
                    .notifying(record -> {
                        log.info("original change record:{}, \n\n{}, \n\n{}",
                                record.key(), record.destination(), record.value());
                        receiveChangeEvent(record.value(), props.getProperty("debezium.name"));
                    }).build());
        }
    }

    private void receiveChangeEvent(String value, String name) {
        if (Objects.nonNull(value)) {
            Map<String, Object> payload = JSONUtils.getPayload(value);
            String handleType = JSONUtils.getHandleType(payload);
            if (!("NONE".equals(handleType) || "READ".equals(handleType))) {
                ChangeRow changeData = JSONUtils.getChangeData(payload);
                Map<String, Object> data;
                if ("DELETE".equals(handleType)) {
                    data = changeData.getBefore();
                } else {
                    data = changeData.getAfter();
                }
                Message build = new Message()
                        .setData(data)
                        .setDbType("Mysql")
                        .setDatabase(String.valueOf(changeData.getSource().get("db")))
                        .setTable(String.valueOf(changeData.getSource().get("table")))
                        .setHandleType(handleType);
                log.info("【Debezium-" + name + "】" + build.toString());
            }
        }
    }

    @Override
    public void destroy() throws Exception {
        for (DebeziumEngine<ChangeEvent<String, String>> engine : engineList) {
            if (Objects.nonNull(engine)) {
                try {
                    engine.close();
                } catch (Exception e) {
                    log.info("close debezium engine error", e);
                }
            }
        }
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        for (DebeziumEngine<ChangeEvent<String, String>> engine : engineList) {
            Executors.newSingleThreadExecutor().execute(engine);
        }
    }
}
